package cn.dmp.image

import cn.dmp.util.{AppParams, RedisHander}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
  * 数据标签化
  */
object KpiImageMainService {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setAppName("数据标签化")
    conf.setMaster("local[*]")
    //序列化方式
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc: SparkContext = new SparkContext(conf)

    //使用广播变量 或者 使用 后续直接从redis获取
    val stopWords: RDD[String] = sc.textFile(AppParams.stopWordsPath)
    val sotopWordBT: Broadcast[Array[String]] = sc.broadcast(stopWords.collect())

    val sQLContext = new SQLContext(sc)

    /**
      * adspacetype: Int, 广告位类型（1：banner 2：插屏 3：全屏）
      * adspacetypename: String, 广告位类型名称（banner、插屏、全屏
      * appid: String, 应用 id
      * appname: String, 应用名称
      * adplatformproviderid: Int, 	广告平台商id  (>= 100000: rtb)
      *
      * client: Int, 操作系统 （1：android 2：ios 3：wp）
      * networkmannername:String  联网方式名称
      * ispname: String, 运营商名称
      * keywords: String, 	关键字
      * provincename: String, 	设备所在省份名称
      * cityname: String, 	设备所在城市名称
      *
      * userId
      *
      * imei: String, 	imei
      * mac: String, 	mac
      * idfa: String, 	idfa
      * openudid: String, 	openudid
      * androidid: String, 	androidid
      * imeimd5: String, 	imei_md5
      * macmd5: String, 	mac_md5
      * idfamd5: String, 	idfa_md5
      * openudidmd5: String, 	openudid_md5
      * androididmd5: String,	androidid_md5
      * imeisha1: String, 	imei_sha1
      * macsha1: String, 	mac_sha1
      * idfasha1: String, 	idfa_sha1
      * openudidsha1: String, 	openudid_sha1
      * androididsha1: String,	androidid_sha1
      *
      */
    //联网方式
    //var clientMap = Map[Int, String](1 -> "D00010001", 2 -> "D00010002", 3 -> "D00010003")
    var clientMap: Map[String, AnyRef] = AppParams.clientMap

    //设备联网方式
    //val networkMap: Map[String, String] = Map[String, String]("WIFI" -> "D00020001", "4G" -> "D00020002", "3G" -> "D00020003", "2G" -> "D00020004")
    var networkMap: Map[String, AnyRef] = AppParams.networkMap

    //设备运营商方式
    //val ispnameMap: Map[String, String] = Map[String, String]("移动" -> "D00030001", "联通" -> "D00030002", "电信" -> "D00030003")
    var ispnameMap: Map[String, AnyRef] = AppParams.ispnameMap

    //读取parquet文件
    val frame: DataFrame = sQLContext.read.parquet(AppParams.parquetOutPath)

    val logMap: RDD[(String, String, String, String, String, String, List[String], String, String)] = frame.map(row => {
      KpiServiceMangerHander.accUserImageMapInfo(row, clientMap, networkMap, ispnameMap, sotopWordBT)
    })

    //方法一
    KpiServiceMangerHander.accKpiImageInfo(logMap)

    //方法二
    KpiServiceMangerHander.accKpiImageInfoRedis(logMap)

    sc.stop()
  }

}
